package com.clp.inflxudb.spring;

import com.clp.inflxudb.core.Flux;
import com.clp.inflxudb.core.FluxHandler;
import com.clp.inflxudb.core.domain.FluxKV;
import com.clp.inflxudb.core.domain.FluxKVGroup;
import com.clp.inflxudb.core.domain.FluxPoint;
import com.clp.inflxudb.core.domain.TsRange;
import com.clp.inflxudb.core.enums.CompareOp;
import com.clp.inflxudb.core.enums.FluxKVType;
import com.clp.inflxudb.core.enums.FluxKey;
import com.clp.inflxudb.core.flux_function.FluxFunction;
import com.clp.inflxudb.core.flux_function.ResultFluxFunction;
import com.influxdb.client.*;
import com.influxdb.client.domain.Bucket;
import com.influxdb.client.domain.BucketRetentionRules;
import com.influxdb.client.domain.Organization;
import com.influxdb.client.domain.WritePrecision;
import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Instant;
import java.util.*;

/**
 * influxdb 的 数据访问类，提供数据访问
 * 对 InfluxDBClient的重新封装，可以为不同Organization的InfluxService提供数据访问
 */
public class FluxTemplate implements AutoCloseable {
    private static final Logger log = LoggerFactory.getLogger(FluxTemplate.class);

    private static class Key {
        private String url;
        private String token;

        public Key(String url, String token) {
            this.url = url;
            this.token = token;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            Key key = (Key) o;
            return url.equals(key.url) && token.equals(key.token);
        }

        @Override
        public int hashCode() {
            return Objects.hash(url, token);
        }
    }

    private static final Map<Key, FluxTemplate> map = new Hashtable<>();

    static FluxTemplate get(String url, String token) {
        FluxTemplate fluxTemplate = map.get(new Key(url, token));
        if (fluxTemplate != null) {
            return fluxTemplate;
        }
        synchronized (FluxTemplate.class) {
            fluxTemplate = map.get(new Key(url, token));
            if (fluxTemplate != null) {
                return fluxTemplate;
            }
            fluxTemplate = new FluxTemplate(url, token);
            map.put(new Key(url, token), fluxTemplate);
        }
        return fluxTemplate;
    }

    private final String urlStr;
    private final String tokenStr;

    private final InfluxDBClient client; // influxdbClient客户端
    private final OrganizationsApi organizationsApi;
    private final BucketsApi bucketsApi;
    private final WriteApi writeApi;
    private final WriteApiBlocking writeApiBlocking;
    private final QueryApi queryApi;
    private final DeleteApi deleteApi;

    // 映射： orgName -> organization对象
    private final Map<String, Organization> orgMap;
    // 映射： orgName -> 该orgName下的所有Map(bucketName -> bucket)
    private final Map<String, Map<String, Bucket>> orgBucketsMap;

    private FluxTemplate(String url, String token) {
        urlStr = url;
        tokenStr = token;
        // 设置客户端
        client = InfluxDBClientFactory.create(urlStr, tokenStr.toCharArray());
        // 设置所有API
        organizationsApi = client.getOrganizationsApi();
        bucketsApi = client.getBucketsApi();
        writeApi = client.getWriteApi();
        writeApiBlocking = client.getWriteApiBlocking();
        queryApi = client.getQueryApi();

        deleteApi = client.getDeleteApi();
        // 设置组织Map
        orgMap = new HashMap<String, Organization>();
        List<Organization> orgList = organizationsApi.findOrganizations();
        orgList.forEach((organization -> orgMap.put(organization.getName(), organization)));
        // 获取该Organization下的 所有 Bucket （相当于MySql的database）Map
        orgBucketsMap = new HashMap<>();
        for (String orgName : orgMap.keySet()) {
            Map<String, Bucket> bucketMap = new HashMap<>();
            for (Bucket bucket : bucketsApi.findBucketsByOrgName(orgName)) {
                bucketMap.put(bucket.getName(), bucket);
            }
            orgBucketsMap.put(orgName, bucketMap);
        }
    }

    /**
     * 判断该InfluxDB数据库是否有该组织
     *
     * @param orgName
     * @return
     */
    private boolean hasOrgName(String orgName) {
        if (orgMap.containsKey(orgName)) {
            return true;
        }
        return false;
    }

    /**
     * 向该Organization下添加新的Bucket
     *
     * @param orgName：组织名称
     * @param bucketName：组织下的Bucket名称
     * @param rules：规则
     */
    public boolean createBucketIfNotExist(String orgName, String bucketName, BucketRetentionRules rules) {
        Map<String, Bucket> bucketMap = orgBucketsMap.get(orgName);
        if (bucketMap == null) {
            return false;
        }
        if (!bucketMap.containsKey(bucketName)) {
            Bucket bucket = bucketsApi.createBucket(bucketName, rules, orgMap.get(orgName).getId());
            bucketMap.put(bucketName, bucket);
        }
        return true;
    }

    public boolean deleteBucketIfExist(String orgName, String bucketName) {
        Map<String, Bucket> bucketMap = orgBucketsMap.get(orgName);
        if (bucketMap != null && !bucketMap.isEmpty() && bucketMap.keySet().contains(bucketName)) {
            bucketsApi.deleteBucket(bucketMap.get(bucketName));
            bucketMap.remove(bucketName);
            return true;
        }
        return false;
    }

//    public boolean deleteBucketMeasurementIfExist(String orgName, String bucketName, String measurementName) {
//        Map<String, Bucket> bucketMap = orgBucketsMap.get(orgName);
//        if (bucketMap != null && !bucketMap.isEmpty() && bucketMap.keySet().contains(bucketName)) {
//            bucketsApi.d
//        }
//    }

    /**
     * 写单个数据到某个bucket
     *
     * @param orgName                    ：组织名称
     * @param bucketName：组织下的bucket（表）名称
     * @param fluxPoint：数据点
     */
    public void writeByFluxPoint(String orgName, String bucketName, FluxPoint fluxPoint) {
        if (!hasOrgName(orgName)) {
            return;
        }
        writeApi.writeMeasurement(bucketName, orgName, WritePrecision.S, fluxPoint);
    }

    /**
     * 写数据列表到某个bucket。根据@Measurement注解会自动写入到不同的measurement
     *
     * @param bucketName：bucket名称
     * @param fluxPointList：数据点列表
     */
    public <T extends FluxPoint> void writeByFluxPoints(String orgName, String bucketName, List<T> fluxPointList) {
        if (!hasOrgName(orgName)) {
            return;
        }
        writeApi.writeMeasurements(bucketName, orgName, WritePrecision.S, fluxPointList);
    }

    public <T extends FluxPoint> void writeBlockingByFluxPoints(String orgName, String bucketName, List<T> fluxPointList) {
        writeBlockingByFluxPoints(orgName, bucketName, fluxPointList, WritePrecision.S);
    }

    public <T extends FluxPoint> void writeBlockingByFluxPoints(String orgName, String bucketName, List<T> fluxPoints, WritePrecision wp) {
        if (!hasOrgName(orgName)) {
            return;
        }
        writeApiBlocking.writeMeasurements(bucketName, orgName, WritePrecision.S, fluxPoints);
    }

    /**
     * 自定义查询方式：指定组织名称和查询语句即可
     *
     * @param orgName：组织
     * @param queryStr：查询语句
     * @return
     */
    public List<FluxTable> query(String orgName, String queryStr) {
        if (!hasOrgName(orgName)) {
            return null;
        }
        log.info("flux query : {}", queryStr);
        return queryApi.query(queryStr, orgName);
    }

    /**
     * 模板化查询
     *
     * @param orgName：组织
     * @param bucketName：桶名称
     * @param clazz：含有测量名称
     * @param fluxHandler：处理器
     * @param <T>
     * @return
     */
    public <T extends FluxPoint> List<T> queryByFluxHandler(String orgName, String bucketName, Class<T> clazz, FluxHandler<T> fluxHandler) {
        if (orgName == null || !hasOrgName(orgName) || fluxHandler == null) {
            return new ArrayList<>();
        }

        TsRange tsRange = fluxHandler.getRange();
        if (tsRange == null) {
            throw new NullPointerException("时间范围不能为空！");
        }
        // 指定 桶、时间范围、测量
        Flux flux = Flux.getInstance().bucket(bucketName).range(tsRange)
                .filter(FluxKey._measurement, FluxPoint.getMeasurementName(clazz), CompareOp.EQUAL);

        // 收集条件 （tag、field、timestamp）
        T condition0 = fluxHandler.getCondition0();
        FluxKVGroup condition1 = fluxHandler.getCondition1();
        if (condition0 != null) {
            // tag
            condition0.getTagNames().forEach(tagName -> {
                Object tagObj = condition0.get(tagName, Object.class);
                if (tagObj != null) {
                    flux.filter(tagName, tagObj.toString(), CompareOp.EQUAL);
                }
            });
            // 时间戳
            String timestampName = condition0.getTimestampName();
            Instant timeStamp = condition0.get(timestampName, Instant.class);
            if (timeStamp != null) {
                flux.filter(FluxKey._time, timeStamp, CompareOp.EQUAL);
            }
        }
        if (condition1 != null) {
            flux.filter(condition1);
        }
        flux.pivot();
        // field
        List<FluxKV> fieldKVs = new ArrayList<>();
        if (condition0 != null) {
            condition0.getFieldNames().forEach(fieldName -> {
                Object fieldObj = condition0.get(fieldName, Object.class);
                if (fieldObj != null) {
                    fieldKVs.add(new FluxKV(fieldName, fieldObj, CompareOp.EQUAL, FluxKVType.FIELD));
                }
            });
            flux.filterAnd(fieldKVs);
        }

        // pivot()之后 .....
        // 添加处理函数
        FluxFunction function = fluxHandler.getFunction();
        if (function != null) {
            function.doFunction(flux);
        }

        // 执行查询
        List<FluxTable> fluxTableList = query(orgName, flux.yield());
        if (function != null) {
            function.afterQuery(fluxTableList);
        }
        fluxHandler.afterQuery(fluxTableList);

        // 封装数据
        if (function instanceof ResultFluxFunction) {
            return new ArrayList<>();
        } else {
            List<T> beforeList = convertFluxTables(fluxTableList, clazz, fluxHandler.isReverseConvertFluxTable());
            return fluxHandler.afterConvert(beforeList);
        }
    }

    /**
     * 封装查询后的单表数据
     *
     * @param fromList
     * @param clazz
     * @param isReverseConvertFluxTable ：封装表的顺序
     * @param <T>
     * @return
     */
    public static <T extends FluxPoint> List<T> convertFluxTables(List<FluxTable> fromList, Class<T> clazz, boolean isReverseConvertFluxTable) {
        List<T> toList = new ArrayList<>();
        String timestampName = T.getTimestampName(clazz);
        List<String> tagNames = T.getTagNames(clazz);
        List<String> fieldNames = T.getFieldNames(clazz);

        if (!isReverseConvertFluxTable) {
            for (FluxTable table : fromList) {
                for (FluxRecord record : table.getRecords()) {
                    T tElem = FluxPoint.getChildInstance(clazz);
                    Map<String, Object> strObjs = record.getValues();

                    tElem.set(timestampName, strObjs.get(FluxKey._time));

                    for (String tagName : tagNames) {
                        tElem.set(tagName, strObjs.get(tagName));
                    }

                    for (String fieldName : fieldNames) {
                        tElem.set(fieldName, strObjs.get(fieldName));
                    }

                    toList.add(tElem);
                }
            }
        } else {
            for (int i = fromList.size() - 1; i >= 0; i--) {
                FluxTable table = fromList.get(i);
                for (FluxRecord record : table.getRecords()) {
                    T tElem = FluxPoint.getChildInstance(clazz);
                    Map<String, Object> strObjs = record.getValues();

                    tElem.set(timestampName, strObjs.get(FluxKey._time));

                    for (String tagName : tagNames) {
                        tElem.set(tagName, strObjs.get(tagName));
                    }

                    for (String fieldName : fieldNames) {
                        tElem.set(fieldName, strObjs.get(fieldName));
                    }

                    toList.add(tElem);
                }
            }
        }
        return toList;
    }

    @Override
    public void close() {
        if (client != null) {
            client.close();
        }
    }

    @Override
    protected void finalize() throws Throwable {
        super.finalize();
        close();
    }
}
